package com.lrj.datastream.sink;

import com.lrj.datastream.domain.ClickEvent;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author lrj
 * @date 2022/3/30 9:37
 */
public class TestJdbcSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<ClickEvent> stream = env.fromElements(
                new ClickEvent("zhangsan", "./cart", 1000L),
                new ClickEvent("lisi", "./home", 1000L),
                new ClickEvent("wangwu", "./home", 1000L),
                new ClickEvent("zhaoliu", "./product?id=1000", 1000L)
        );
        String sql = "insert into click(userId,url,timestamp) values(?,?,?)";
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String url = parameterTool.get("url");
        String user = parameterTool.get("user", "root");
        String password = parameterTool.get("password");
        String driver=parameterTool.get("driver","com.mysql.cj.jdbc.Driver");
        JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUsername(user)
                .withPassword(password)
                .withDriverName(driver)
                .withUrl(url)
                .build();
        SinkFunction<ClickEvent> sink = JdbcSink.sink(sql, new MyJdbcStatementBuilder(), connectionOptions);
        stream.addSink(sink);
        env.execute(TestJdbcSink.class.getName());
    }

    private static class MyJdbcStatementBuilder implements JdbcStatementBuilder<ClickEvent> {
        @Override
        public void accept(PreparedStatement preparedStatement, ClickEvent event) throws SQLException {
            preparedStatement.setString(1, event.getUser());
            preparedStatement.setString(2, event.getUrl());
            preparedStatement.setLong(3, event.getTimestamp());
        }
    }
}
